package com.hba.consumer;

import com.hba.annotation.QueueAnnotation;
import com.hba.broker.RabbitBroker;
import com.hba.constant.RabbitConstant;
import com.hba.util.PackageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Configuration
public class MessageConsumer extends RabbitBroker{

    private static final Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
    public static ConcurrentMap<String, String> consumerMap=new ConcurrentHashMap<String, String>();

    @Bean
    public SimpleMessageListenerContainer initConstainer(ConnectionFactory connectionFactory,
                                                         MessageListenerAdapter messageListenerAdapter) throws Exception{
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setMessageListener(messageListenerAdapter);
        List<String> classNames =  PackageUtil.getClassName(RabbitConstant.DEFAULT_SCAN_PACKAGE_NAME);
        for(String className : classNames){
            logger.debug("now load class:" + className);
            try{
                if(Class.forName(className).isAnnotationPresent(QueueAnnotation.class)){
                  QueueAnnotation queueAnnotation =  Class.forName(className).getAnnotation(QueueAnnotation.class);
                  String queueName = queueAnnotation.queueName();
                  String exchangeName = queueAnnotation.exchangeName();
                  logger.debug("now load queueName:" + queueName);
                  boolean delay =  queueAnnotation.delay();
                  boolean order =  queueAnnotation.order();
                    StringBuilder queueBuild=new StringBuilder();
                  if(delay){
                      queueBuild.append(exchangeName);
                      declareBindExchange(RabbitConstant.DEFAULT_DELAY_EXCHANGE_NAME,queueName+RabbitConstant.DEFAULT_SEQUEUE_DELAY_SUFFIX,exchangeName,queueName);
                      declareBindExchange(exchangeName,queueName,RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE,queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX);
                      declareDeadLetterBindExchange(RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE,queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX,RabbitConstant.DEFAULT_DELAY_EXCHANGE_NAME,queueName+RabbitConstant.DEFAULT_SEQUEUE_DELAY_SUFFIX);
                  }else{
                      queueBuild.append(exchangeName);
                      declareBindExchange(exchangeName,queueName,RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE,queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX);
                      declareDeadLetterBindExchange(RabbitConstant.DEFAULT_DEAD_LETTER_EXCHANGE,queueName+RabbitConstant.DEFAULT_SEQUEUE_RETRY_SUFFIX,exchangeName,queueName);

                  }
                    queueBuild.append("|");
                    queueBuild.append(queueName);
                    container.setQueueNames(queueName);
                    consumerMap.putIfAbsent(queueBuild.toString(), getBeanName(className));
                    container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
                    logger.debug("cpu thread {}",Runtime.getRuntime().availableProcessors());
                    if(order){
                        container.setConcurrentConsumers(1);
                    }else{
                        container.setConcurrentConsumers(2*Runtime.getRuntime().availableProcessors()-1);
                    }
                    container.setPrefetchCount(1);
                    container.setTxSize(1);
                }
            }catch (Exception e){
                logger.error("class load queue fail:{}",e);
                throw new Exception("class load queue fail:",e);
            }
        }
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(MessageReceiver receiver) {
        return new MessageListenerAdapter(receiver, new SimpleMessageConverter());
    }

    private String getBeanName(String className){
        if(!className.contains(".")){
            logger.error("getBeanName param className error");
            return "";
        }
        String beanNameBig=className.substring(className.lastIndexOf(".")+1);
        char temp=beanNameBig.charAt(0);
        temp=(char) (temp+32);
        return temp+beanNameBig.substring(1, beanNameBig.length());
    }


}
